package cn.gwm.flink.streaming.dws.reservecharge;

import cn.gwm.flink.streaming.client.HbaseClient;
import cn.gwm.flink.streaming.client.RedisClient;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.DefaultConstant;
import cn.gwm.flink.streaming.dwm.charge.model.ChargeStateEnum;
import cn.gwm.flink.streaming.dwm.charge.util.SupplyUtil;
import cn.gwm.flink.streaming.dws.reservecharge.beans.RcElectricConfigDao;
import cn.gwm.flink.streaming.dws.reservecharge.enums.ChargeConnectTypeEnum;
import cn.gwm.flink.streaming.dws.reservecharge.enums.ChargeUseElectricTypeEnum;
import cn.gwm.utils.ConfigLoader;
import cn.gwm.utils.IntegralAreaSoe;
import cn.hutool.core.util.ObjectUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.List;
import java.util.Map;


/**
 * @author Liu.Nanfang
 * @date 2022/7/19 13:06
 * @description
 */
@Slf4j
public class ReserveChargeRichFlatMapFunction extends RichFlatMapFunction<ReserveChargeData, ReserveChargeData> {
    /**
     * 上一条数据key
     */
    private final String PRE_DATA_STATE = "preDataState";
    /**
     * 电价配置信息key
     */
    private final String ELECTRIC_CONFIG_DATA_STATE = "electricConfigState";
    /**
     * 保存充电开始时间状态key
     */
    private final String START_CHARGE_TIME_STATE = "startTime";
    /**
     * soc 缓存key
     */
    private final String ANXIETY_SOC_CACHE_KEY = "charge:dws:anxiety_soc:vin:";
    /**
     * adCode 缓存key
     */
    private final String DWS_CHARGE_AD_CODE_CACHE_KEY = "charge:dws:ad_code:vin:";

    /**
     * soc 缓存时间 1 天
     */
    private final long ANXIETY_SOC_CACHE_KEY_EXPIRE = 1 * 60 * 60 * 24;

    /**
     * adCode 缓存时间 7 天
     */
    private final long AD_CODE_CACHE_KEY_EXPIRE = 7 * 60 * 60 * 24;

    private ChargePrice chargePrice;
    /**
     * 上一条数据
     */
    private ValueState<ReserveChargeData> preDataState;

    private ValueState<Long> busiStartChargeTimeState;

    private ValueState<ReserveChargeElectricConfig> electricConfig;


    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        preDataState = getRuntimeContext().getState(new ValueStateDescriptor<>(PRE_DATA_STATE, ReserveChargeData.class));
        electricConfig = getRuntimeContext().getState(new ValueStateDescriptor<>(ELECTRIC_CONFIG_DATA_STATE, ReserveChargeElectricConfig.class));
        busiStartChargeTimeState = getRuntimeContext().getState(new ValueStateDescriptor<>(START_CHARGE_TIME_STATE, Long.class));
        // 初始化电价信息类
        chargePrice = ChargePrice.getInstance();
        if (chargePrice.getChargePriceInfoList().size() == 0) {
            chargePrice.getChargePriceInfoList();
        }
    }

    @Override
    public void flatMap(ReserveChargeData nowData, Collector<ReserveChargeData> out) throws Exception {
        ReserveChargeData preData = preDataState.value();

        if (ObjectUtil.isNull(preData)) {
            handleGeneralSetting(nowData, null);
            out.collect(nowData);
            return;
        }
        handleGeneralSetting(nowData, preData);
        out.collect(nowData);
        log.info("flatMap env active: {}", System.getenv().get(DefaultConstant.BOOT_PROFILES_ACTIVE));
        log.info("flatMap active: {}", ConfigLoader.get(DefaultConstant.BOOT_PROFILES_ACTIVE_PREFIX));
        log.info("reserveChargeData: vin:{}, chargeStartTime:{}" + nowData.getDeviceId(), nowData.getStartChargeTime());
    }

    /**
     * 处理业务充电开始结束数据
     *
     * @param nowData
     * @param preData
     * @throws IOException
     */
    private void computerChargeTime(ReserveChargeData nowData, ReserveChargeData preData) throws IOException {
        if (preData == null) {
            if (!Long.valueOf(BaseFields.defaultVal).equals(nowData.getStartChargeTime())) {
                nowData.setBusiStartChargeTime(nowData.getStartChargeTime());
            }
            if (!Long.valueOf(BaseFields.defaultVal).equals(nowData.getEndChargeTime())) {
                nowData.setBusiEndChargeTime(nowData.getEndChargeTime());
            }
        } else {
            if (ObjectUtil.isNotNull(nowData.getEndFlag()) && ChargeStateEnum.END.getCode().equals(nowData.getChargeStatus())) {
                busiStartChargeTimeState.clear();
                nowData.setBusiStartChargeTime(nowData.getStartChargeTime());
                nowData.setBusiEndChargeTime(nowData.getEndChargeTime());
//                if (Integer.valueOf(ChargeModelEnum.CHARGE_MULTIMODE.getCode()).equals(nowData.getChargeModel()) &&
//                        nowData.getDiscountPower() != null &&
//                        nowData.getDiscountPower().compareTo(0.0) > 0) {
//                    nowData.setFavourable(Boolean.TRUE);
//                }

                log.info("reserveChargeData clear vin:{} nowTime:{} preTime:{}", nowData.getDeviceId(), nowData.getEndChargeTime(), nowData.getEndChargeTime());
                preDataState.clear();
            } else {
                if (ChargeStateEnum.CHARGING.getCode().equals(nowData.getChargeStatus()) &&
                        !Long.valueOf(BaseFields.defaultVal).equals(nowData.getStartChargeTime()) &&
                        ObjectUtil.isNull(busiStartChargeTimeState.value())
                ) {
                    nowData.setBusiStartChargeTime(nowData.getStartChargeTime());
                    busiStartChargeTimeState.update(nowData.getStartChargeTime());
                    log.info("update time: status:{} vin:{} date:{}", nowData.getChargeStatus(), nowData.getDeviceId(), nowData.getStartChargeTime());
                } else if (!ChargeStateEnum.WAIT.getCode().equals(nowData.getChargeStatus()) &&
                        ObjectUtil.isNotNull(busiStartChargeTimeState.value())) {
                    nowData.setBusiStartChargeTime(busiStartChargeTimeState.value());
                    log.info("set update time: status:{} vin:{} date:{}", nowData.getChargeStatus(), nowData.getDeviceId(), nowData.getStartChargeTime());
                }
                // 处理充电结束时间 ,当充满电情况，提取充电结束时间
                if (ObjectUtil.isNull(nowData.getEndFlag()) && ChargeStateEnum.FULLY.getCode().equals(nowData.getChargeStatus())) {
                    nowData.setBusiEndChargeTime(nowData.getEndChargeTime());

                }

            }
        }

        if (Long.valueOf(BaseFields.defaultVal).equals(nowData.getEndChargeTime())) {
            nowData.setEndChargeTime(null);
        }
        if (Long.valueOf(BaseFields.defaultVal).equals(nowData.getEnd_chrgConnectTid())) {
            nowData.setEnd_chrgConnectTid(null);
        }

    }

    /**
     * 处理状态数据
     *
     * @param nowData
     * @param preData
     * @throws IOException
     */
    private void handleChargeState(ReserveChargeData nowData, ReserveChargeData preData) throws IOException {
        if (preData == null) {
            if (!ChargeStateEnum.END.getCode().equals(nowData.getChargeStatus())) {
                preDataState.update(nowData);
            }
        } else {
            if (ObjectUtil.isNotNull(nowData.getEndFlag()) && ChargeStateEnum.END.getCode().equals(nowData.getChargeStatus())) {
                preDataState.clear();
            }
        }
    }

    /**
     * 在状态管理没有数据情况下
     *
     * @param nowData
     * @throws IOException
     */
    private void handleFirstData(ReserveChargeData nowData) {
        // 判断经纬度是否存在
        findAndReplaceAdCode(nowData);
        updateElectricConfigByAdCode(nowData.getAdCode());
    }

    /**
     * 根据adCode 设置电价相关信息
     *
     * @param adCode
     */
    private void updateElectricConfigByAdCode(String adCode) {
        if (ObjectUtil.isNull(adCode) || "null".equals(adCode) || "[]".equals(adCode) ||
                BaseFields.defaultVal.equals(adCode)) {
            return;
        }
        try {
            RcElectricConfigDao filterElectricConfig = chargePrice.getFilterByAdCode(adCode, ChargeUseElectricTypeEnum.RESIDENTS.getCode());
            if (filterElectricConfig != null) {
                ReserveChargeElectricConfig electricConfigDao = chargePrice.getReserveChargeElectricConfig(filterElectricConfig);
                electricConfig.update(electricConfigDao);
            } else {
                electricConfig.clear();
            }


        } catch (IOException e) {
            e.printStackTrace();
            log.info("updateElectricConfig err: {}", e.toString());
        }


    }


    /**
     * 更新数据
     *
     * @param preData
     * @param nowData
     * @throws Exception
     */
    private void handleUpdateData(ReserveChargeData nowData, ReserveChargeData preData) {
        if (preData == null) {
            return;
        }
        String preAdCode = preData.getAdCode();
        if (ObjectUtil.isNotNull(preAdCode) &&
                (!"[]".equals(preAdCode) && !BaseFields.defaultVal.equals(preAdCode))) {
            nowData.setAdCode(preData.getAdCode());
            nowData.setLatitude(preData.getLatitude());
            nowData.setLongitude(preData.getLongitude());
        } else {
            handleFirstData(nowData);
        }

//        nowData.setCalculateChargingCapacity(preData.getCalculateChargingCapacity());
        if (preData.getBusiEndChargeTime() != null) {
            nowData.setBusiEndChargeTime(preData.getBusiEndChargeTime());
        }
    }


    /**
     * 设置通用信息
     *
     * @param nowData
     * @param preData
     */
    private void handleGeneralSetting(ReserveChargeData nowData, ReserveChargeData preData) throws IOException {
        if (ObjectUtil.isNull(preData)) {
            handleFirstData(nowData);
        }
        // 更新数据
        handleUpdateData(nowData, preData);
        // 处理充电时间
        computerChargeTime(nowData, preData);
        try {
            ReserveChargeElectricConfig electricConfigDao = electricConfig.value();
            // 处理类型
            BuChargeHandle.chargeTypeHandle(nowData, electricConfigDao);
        } catch (Exception e) {
            log.error("electricConfig fail e: {}", e);
        }
        // 设置电价信息
        setElectricPrice(nowData);
        // 业务充电状态合并
        ChargeStateEnum stateEnum = ChargeStateEnum.fromKey(nowData.getChargeStatus());
        nowData.setBuChargeStatus(BuChargeHandle.handleBuChargeStatus(stateEnum));
        // 只有测试环境可用
        if (DefaultConstant.BOOT_TEST_ACTIVE.equals(ConfigLoader.get(DefaultConstant.BOOT_PROFILES_ACTIVE_PREFIX))) {
            nowData.setAnxietySoc(getAnxietySoc(nowData.getDeviceId()));
        }
        if (ChargeConnectTypeEnum.SLOW.getCode().equals(nowData.getChargeConnectType())) {
            // 目前只预测慢充情况下的充电时长
            Long calculateChargeTime = getCalculateChargeTime(nowData.getDeviceId(), nowData.getStartChargeTime());
            if (calculateChargeTime != null) {
                nowData.setCalculateChargeTimeMinute(calculateChargeTime);
            }
        }
        // 计算电价信息
        setCalculHandlePowerAndPrice(nowData, preData);
        ChargePrice.computerChargePowerAmount(nowData);
        handleChargeState(nowData, preData);
        setCalculationSoeHandle(nowData);
    }

    /**
     * 查找并替换 adcode
     *
     * @param nowData
     */
    private void findAndReplaceAdCode(ReserveChargeData nowData) {
        String nowAdCode = nowData.getAdCode();
        try {
            if (ObjectUtil.isNotNull(nowAdCode) &&
                    (!"[]".equals(nowAdCode) && !BaseFields.defaultVal.equals(nowAdCode))) {
                setAdCodeByRedis(nowData.getDeviceId(), nowAdCode);
                return;
            }
            if (ObjectUtil.isNotNull(nowData.getLatitude()) && ObjectUtil.isNotNull(nowData.getLongitude())) {
                if (BaseFields.defaultVal.equals(nowData.getLatitude().toString()) ||
                        BaseFields.defaultVal.equals(nowData.getLongitude().toString())) {
                    String adCode = getAdCodeByRedis(nowData.getDeviceId());
                    if (adCode != null) {
                        nowData.setAdCode(adCode);
                    }
                } else {
                    Map<String, String> code = SupplyUtil.supplyCode(nowData.getLatitude(), nowData.getLongitude());
                    if (ObjectUtil.isNotEmpty(code) && ObjectUtil.isNotNull(code.get("adCode")) && !"[]".equals(nowAdCode)) {
                        nowData.setAdCode(code.get("adCode"));
                        if (StringUtils.isNotBlank(code.get("aoiType"))) {
                            nowData.setAoiType(code.get("aoiType"));
                        }
                        setAdCodeByRedis(nowData.getDeviceId(), code.get("adCode"));
                    }
                }
            } else {
                nowData.setAdCode(getAdCodeByRedis(nowData.getDeviceId()));
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("find or replace adCode err : {}", e);
        }
    }

    /**
     * 测试，随机生成电量信息 计算
     *
     * @param reserveChargeData
     * @param pre
     */
    private void setCalculHandlePowerAndPrice(ReserveChargeData reserveChargeData, ReserveChargeData pre) {
        // 计算电价

        if (reserveChargeData.getDiscountPower() != null && !Integer.valueOf(BaseFields.defaultVal).equals(reserveChargeData.getDiscountPower()) &&
                reserveChargeData.getExpensivePower() != null && !Integer.valueOf(BaseFields.defaultVal).equals(reserveChargeData.getExpensivePower())) {
            reserveChargeData.setChargingCapacity(reserveChargeData.getDiscountPower() + reserveChargeData.getExpensivePower());
        }

        if (Integer.valueOf(BaseFields.defaultVal).equals(reserveChargeData.getDiscountPower())) {
            reserveChargeData.setDiscountPower(null);
        }
        if (Integer.valueOf(BaseFields.defaultVal).equals(reserveChargeData.getExpensivePower())) {
            reserveChargeData.setExpensivePower(null);
        }

    }

    /**
     * 计算soe
     *
     * @param nowData
     */
    public void setCalculationSoeHandle(ReserveChargeData nowData) {
        if (nowData.getBMS_InnerSOC() != null && !(new BigDecimal(BaseFields.defaultVal)).equals(nowData.getBMS_InnerSOC())) {
            nowData.setBMS_SOE(IntegralAreaSoe.computeSoe(nowData.getBMS_InnerSOC().doubleValue()));
        }
        if (nowData.getEnd_BMS_InnerSOC() != null && !(new BigDecimal(BaseFields.defaultVal)).equals(nowData.getEnd_BMS_InnerSOC())) {
            nowData.setEnd_BMS_SOE(IntegralAreaSoe.computeSoe(nowData.getEnd_BMS_InnerSOC().doubleValue()));
        }
    }


    /**
     * 设置电价信息
     *
     * @param reserveChargeData
     */
    private void setElectricPrice(ReserveChargeData reserveChargeData) {
        try {
            ReserveChargeElectricConfig electricConfigDao = electricConfig.value();
            if (electricConfigDao != null) {
                ChargePrice.setElectricPrice(electricConfigDao, reserveChargeData);
            }
            // 电价获取原则：取第二阶梯 低谷与不分时的价格差价 与优惠电量的乘积
        } catch (Exception e) {
            e.printStackTrace();
            log.info("setElectricPrice err: {}", e.toString());
        }

    }

    /**
     * 设置地区编码至 hbase
     *
     * @param reserveChargeData
     * @throws Exception
     */
    private void putHbaseAdCode(ReserveChargeData reserveChargeData) throws Exception {
        if (reserveChargeData.getAdCode() != null) {
            HbaseClient.putData("DWS_RESERVE_CHARGE_AD_CODE", reserveChargeData.getDeviceId(), "C1", "adCode", String.valueOf(reserveChargeData.getAdCode()));
        }
    }

    /**
     * 根据vin从 hbase 获取地区编码
     *
     * @param vin
     * @return
     * @throws IOException
     */
    private String getHbaseAdCode(String vin) throws IOException {
        List<Map.Entry> list = HbaseClient.getRow("DWS_RESERVE_CHARGE_AD_CODE", vin);
        if (list.size() > 0) {
            return (String) list.get(0).getValue();
        }
        return null;
    }

    private void setAdCodeByRedis(String vin, String adCode) {
        try {
            RedisClient.setData(DWS_CHARGE_AD_CODE_CACHE_KEY + vin, adCode, AD_CODE_CACHE_KEY_EXPIRE);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("setAdCodeByRedis fail :" + e);
        }

    }

    private String getAdCodeByRedis(String vin) {
        try {
            String adCode = RedisClient.getData(DWS_CHARGE_AD_CODE_CACHE_KEY + vin);
            if (StringUtils.isNotBlank(adCode)) {
                return adCode;
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("getAdCodeByRedis fail :" + e);
        }
        return null;

    }


    /**
     * 根据vin 获取焦虑soc 优先从缓存获取
     *
     * @param vin
     * @return
     */
    public Integer getAnxietySoc(String vin) {
        try {
            String soc = RedisClient.getData(ANXIETY_SOC_CACHE_KEY + vin);
            if (StringUtils.isNotBlank(soc)) {
                return Integer.valueOf(soc);
            }
            Integer anxietySoc = SupplyUtil.getAnxietySoc(vin);
            if (anxietySoc != null) {
                RedisClient.setData(ANXIETY_SOC_CACHE_KEY + vin, String.valueOf(anxietySoc), ANXIETY_SOC_CACHE_KEY_EXPIRE);
                return anxietySoc;
            }
        } catch (Exception e) {
            log.error("get anxietySco fail :" + e);
        }
        return null;
    }

    /**
     * 根据vin 充电开始时间获取 充电预测时间（分钟）
     *
     * @param vin
     * @param chargeStartTime
     * @return
     */
    public Long getCalculateChargeTime(String vin, Long chargeStartTime) {
        if (chargeStartTime == null) {
            return null;
        }
        try {
            String key = String.format("charge:status:%s:%s", vin, chargeStartTime);
            String result = RedisClient.getData(key);
            if (StringUtils.isNotBlank(result)) {
                return Long.valueOf(result);
            }
        } catch (Exception e) {
            log.error("get CalculateChargeTime fail :" + e);
        }
        return null;

    }

}
